package com.zr.eim.kafka;

import com.zr.eim.domain.EimDataNucleicAcidCheck;
import com.zr.eim.service.IEimDataNucleicAcidCheckService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.text.ParseException;
import java.text.SimpleDateFormat;

@Component
public class NucleicAcidCheckListener {

    @Resource
    private IEimDataNucleicAcidCheckService service;
    SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

//    @KafkaListener(topics = "NACheckInfo")
    public void listen(ConsumerRecord<String, String> record, Acknowledgment ack){

        String value = record.value();

        System.out.println(value);
        System.out.println(record);

        String [] sa = value.split(",");

        if(sa.length<10)
            return;

        EimDataNucleicAcidCheck info = new EimDataNucleicAcidCheck();
        info.setId(Long.valueOf(sa[0]));
        info.setPersonId(sa[1]);
        info.setInspectionOrg(sa[2]);
        info.setAcquisitionOrg((sa[3]));
        try {
            info.setInspectionTime(sf.parse(sa[4]));
            info.setAcquisitionTime(sf.parse(sa[5]));
        } catch (ParseException e) {
            throw new RuntimeException(e);
        }
        info.setFinish(Long.parseLong(sa[6]));
        info.setResult(Long.parseLong(sa[7]));
        info.setType(sa[8]);
        info.setMode(sa[9]);

        service.addNucleicAcidCheck(info);

        //手动提交offset
        ack.acknowledge();
    }

}
